use bytes::Bytes;
use futures::channel::oneshot;
use hashbrown::HashMap;
use xactor::{Actor, Addr, AnyHandler, Message};

use crate::message::{HoneybeeMessage, RpcResponse, RpcResponseError, RpcResult};

type Response = Box<dyn FnOnce(Bytes) + Send>;

pub struct RpcResponseRouter(HashMap<u32, Response>);

impl RpcResponseRouter {
    pub fn new() -> Self {
        Self(HashMap::default())
    }

    pub fn register<T, R>(&mut self, session: u32, tx: oneshot::Sender<Result<R, RpcResponseError>>)
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let r = self.0.insert(
            session,
            Box::new(move |bytes| {
                let response: Result<RpcResponse<T::Result>, _> =
                    bincode::deserialize(bytes.as_ref());
                if let Ok(response) = response {
                    if let Err(_err) = tx.send(response.0) {
                        //error!("err:{:?}", err);
                    }
                }
            }),
        );
        if r.is_some() {
            error!("rpc session already register !!!");
        }
    }

    pub fn register_async<A, T, R>(&mut self, session: u32, source: Addr<A>)
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let r = self.0.insert(
            session,
            Box::new(move |bytes| {
                let response: Result<RpcResponse<T::Result>, _> =
                    bincode::deserialize(bytes.as_ref());
                if let Ok(response) = response {
                    match response.0 {
                        Ok(res) => drop(source.any_send(res)),
                        Err(err) => error!("async call err:{}", err),
                    }
                }
            }),
        );
        if r.is_some() {
            error!("rpc session already register !!!");
        }
    }

    pub fn response(&mut self, session: u32, bytes: Bytes) {
        if let Some(response) = self.0.remove(&session) {
            response(bytes);
        }
    }
}
